package client

import (
	"errors"
	"gitee.com/cooge/impush/config"
	"gitee.com/cooge/impush/message"
	"gitee.com/cooge/impush/stream"
	"gitee.com/cooge/impush/util"
	log "github.com/sirupsen/logrus"
	"net"
	"strconv"
	"strings"
	"time"
)

const (
	START = 0
	RUN   = 1
	END   = -1
)

type ClusterClient struct {
	ms                  *stream.MessageStream
	host                string
	port                int
	Raddress            string
	handler             HandlerFunc
	messageHandler      MessageHandlerFunc
	sendUserHandlerFunc SendUserHandlerFunc
	reTryNum            int
	Status              int
}

func NewClusterClient(host string, port int) *ClusterClient {
	var cc ClusterClient
	cc.host = host
	cc.port = port
	cc.reTryNum = 5
	cc.Status = START
	cc.Raddress = util.Merge(cc.host, ":", strconv.Itoa(cc.port))
	return &cc
}

func NewClusterClient2(addr string) *ClusterClient {
	var cc ClusterClient
	ads := strings.Split(addr, ":")
	cc.host = ads[0]
	p, err := strconv.Atoi(ads[1])
	if err == nil {
		cc.port = p
	}
	cc.reTryNum = 5
	cc.Raddress = addr
	return &cc
}

type HandlerFunc func(*ClusterClient, int)

type SendUserHandlerFunc func(*ClusterClient, string, int)

type MessageHandlerFunc func(*ClusterClient, *message.Message, string) error

func (this *ClusterClient) Start(handler HandlerFunc, messageHandler MessageHandlerFunc, sendUserHandlerFunc SendUserHandlerFunc) {
	this.handler = handler
	this.messageHandler = messageHandler
	this.sendUserHandlerFunc = sendUserHandlerFunc
	this.xstart()
}

func (this *ClusterClient) xstart() {
	var times = 0
	for {
		times++

		log.Info("尝试连接集群服务器", this.Raddress, "中", times)

		err := this.start()
		if err == nil {
			laddr := this.ms.Conn.LocalAddr().(*net.TCPAddr)
			if laddr.IP.String() == this.host && config.CLUSTER_LOCAL_PORT == this.port {
				log.Info("连接到自身强行关闭", this.Raddress)
				this.reTryNum = 0
				this.close()
				break
			} else {
				this.Status = RUN
				this.handler(this, config.CLUSTER_CLIENT_START_SUCCESS)
				log.Info("连接集群服务器", this.Raddress, "成功", times)
				break
			}
		} else {
			log.Info("尝试连接集群服务器", this.Raddress, "次失败", times)
		}
		if times >= this.reTryNum {
			log.Info("尝试连接集群服务器", this.Raddress, "失败", times)
			this.Status = END
			this.handler(this, config.CLUSTER_CLIENT_START_ERROR)
			break
		}
		time.Sleep(time.Duration(3) * time.Second)
	}
}

func (this *ClusterClient) close() {
	this.ms.Conn.Close()
}

func (this *ClusterClient) start() error {
	tcpAddr, err := net.ResolveTCPAddr("tcp4", this.Raddress)
	if err == nil {
		conn, err := net.DialTCP("tcp4", nil, tcpAddr)
		if err == nil {
			this.ms = stream.NewMessageStream(conn)
			go this.read()
			return nil
		} else {
			log.Info("conn remote  error", this.Raddress, err.Error())
		}
	} else {
		log.Info("remote address error", this.Raddress)
	}
	return errors.New("START FAIL")
}

func (this *ClusterClient) write(msg []byte) (int, error) {
	if this.Status == RUN {
		num, err := this.ms.Conn.Write(msg)
		return num, err
	} else {
		return 0, nil
	}
}
func (this *ClusterClient) Write(msg []byte) (int, error) {
	return this.write(msg)
}

func (this *ClusterClient) Live(address []string) (int, error) {
	address = util.RemoveArrays(address, this.Raddress)
	addr, ok := this.ms.Conn.LocalAddr().(*net.TCPAddr)
	if ok {
		address = append(address, util.Merge(addr.IP.String(), ":", strconv.Itoa(config.CLUSTER_LOCAL_PORT)))
		msg := message.SendClusterLiveMessage(address)
		return this.Write(msg)
	}
	return 0, errors.New("LOCAL ADDRSSS ERROR")
}

func (this *ClusterClient) read() {
	for {
		rms, err := this.ms.ReadMessage()
		if err == nil {
			this.handle(rms)
		} else {
			log.Debug("remote conn break", this.Raddress)
			break
		}
	}
	this.ms.Conn.Close()
	this.Status = END
	if this.reTryNum > 0 {
		go this.xstart()
	} else {
		this.handler(this, config.CLUSTER_CLIENT_NET_BREAK)
	}

}

func (this *ClusterClient) handle(msg *message.Message) {
	ty := msg.Heads.Get(message.TYPE)
	switch ty {
	case message.CLUSTER:
		{
			action := msg.Heads.Get(message.ACTION)
			switch action {

			case message.LIVE:
				{
					log.Debug("收到集群服务端的心跳回馈")
				}
			}
		}
	case message.MESSAGE:
		{
			touser := msg.Heads.Get(message.TO)
			err := this.messageHandler(this, msg, touser)
			if err != nil {
				log.Debug(touser, err)
				this.Write(message.SendMachinesNoUser(touser))
			}
		}
	case message.ALLMESSAGE:
		{

			msg.Heads[message.TYPE] = message.MESSAGE
			touser := msg.Heads.Get(message.TO)
			err := this.messageHandler(this, msg, touser)
			if err != nil {
				log.Debug(touser, err)
				this.Write(message.SendMachinesNoUser(touser))
			} else {
				log.Debug("同步用户", touser)
				this.sendUserHandlerFunc(this, touser, 0)
			}

		}

	}
}
